#!/usr/bin/python3 -cimport os, sys; os.execv(os.path.dirname(sys.argv[1]) + "/../common/pywrap", sys.argv)

# This file is part of Cockpit.
#
# Copyright (C) 2021 Red Hat, Inc.
#
# Cockpit is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.
#
# Cockpit is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Cockpit; If not, see <https://www.gnu.org/licenses/>.

import subprocess

import packagelib
import storagelib
import testlib

PV_SIZE = 4000  # 4 GB in MB


def get_stratis_stop_type_opt(execute):
    """Get `stratis stop pool` required option for a pool name

    The CLI changed in an incompatible way in Fedora 40, it needs an extra --name option
    which cannot be provided in earlier versions.
    """
    try:
        if '--name' in execute("stratis pool stop --help"):
            return "--name"
    except subprocess.CalledProcessError:
        # on RHEL 8 this fails with "error: invalid choice"
        pass
    return ""


def create_pool_key(machine, keyname, passphrase):
    # this is a bit complicated, see https://bugzilla.redhat.com/show_bug.cgi?id=2246923
    machine.execute(f"echo -n '{passphrase}' | stratis key set --keyfile-path /dev/stdin {keyname}")


@testlib.skipImage("No Stratis", "debian-*", "ubuntu-*")
@testlib.skipImage("commit 817c957899a4 removed Statis 2 support", "rhel-8-*")
@testlib.nondestructive
class TestStorageStratis(storagelib.StorageCase):
    def setUp(self):
        super().setUp()
        exe = self.machine.execute

        exe("systemctl start stratisd")
        self.addCleanup(exe, "systemctl stop stratisd")

        self.stop_type_opt = get_stratis_stop_type_opt(exe)

        self.addCleanup(exe,
                        "stratis report | jq -r '.pools[] | .name' |"
                        "xargs -n1 --no-run-if-empty stratis pool destroy")
        self.addCleanup(exe,
                        "stratis report | jq -r '.pools[] | .name' |"
                        f"xargs -n1 --no-run-if-empty stratis pool stop {self.stop_type_opt}")
        self.addCleanup(exe,
                        "mount | grep mapper/stratis | awk '{print $1}' | xargs --no-run-if-empty umount")

    def testBasic(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        # use fixed names, to avoid grabbing loop1 and loop12 (which losetup sometimes likes to do)
        # as that clashes with :contains(loop1) below; also, fix names for pixel test
        dev_1 = self.add_loopback_disk(PV_SIZE, name="loop10")
        dev_2 = self.add_loopback_disk(PV_SIZE, name="loop11")
        dev_3 = self.add_loopback_disk(PV_SIZE, name="loop12")
        dev_4 = self.add_loopback_disk(PV_SIZE, name="loop13")
        dev_5 = self.add_loopback_disk(PV_SIZE, name="loop14")
        b.wait_visible(self.card_row("Storage", name=dev_1))
        b.wait_visible(self.card_row("Storage", name=dev_2))
        b.wait_visible(self.card_row("Storage", name=dev_3))
        b.wait_visible(self.card_row("Storage", name=dev_4))
        b.wait_visible(self.card_row("Storage", name=dev_5))

        # Create a pool
        self.dialog_open_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                    expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                                    self.dialog_is_present('disks', dev_2) and
                                                    self.dialog_check({"name": "pool0"})))
        self.dialog_set_val("disks", {dev_1: True, dev_2: True})
        b.assert_pixels("#dialog", "create-pool")
        self.dialog_apply()
        self.dialog_wait_close()

        b.wait_visible(self.card_row("Storage", name="pool0"))
        b.wait_not_present(self.card_row("Storage", name="pool0") + " .ct-icon-exclamation-triangle")

        # Check that the next name is "pool1"
        self.click_devices_dropdown("Create Stratis pool")
        self.dialog_wait_open()
        self.dialog_wait_val("name", "pool1")
        self.dialog_cancel()
        self.dialog_wait_close()

        # Stop the pool
        pool_uuid = m.execute("stratis --unhyphenated-uuids pool list --name pool0 | grep ^UUID | cut -d' ' -f2").strip()
        m.execute(f"stratis pool stop {self.stop_type_opt} pool0")
        b.wait_in_text(self.card_row("Storage", name=pool_uuid), "Stratis pool")

        # Start it
        self.click_dropdown(self.card_row("Storage", name=pool_uuid), "Start")
        b.wait_visible(self.card_row("Storage", name="pool0"))

        self.click_card_row("Storage", name="pool0")
        b.wait_text(self.card_desc("Stratis pool", "Name"), "pool0")
        b.wait_in_text(self.card_desc("Stratis pool", "Usage"), "8 GB")
        b.wait_not_present('.pf-v5-c-alert')

        udisk_contains_stratis_private = "physical-originsub" in m.execute("udisksctl dump")

        # Create two filesystems
        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_set_val('name', 'fsys1')
        self.dialog_set_val('mount_point', '/run/fsys1')
        b.assert_pixels("#dialog", "create-fsys")
        self.dialog_apply()
        self.dialog_wait_close()
        self.addCleanupMount("/run/fsys1")

        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")

        self.assertEqual(self.inode(m.execute("findmnt -n -o SOURCE /run/fsys1").strip()),
                         self.inode("/dev/stratis/pool0/fsys1"))

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog({'name': 'fsys2',
                     'mount_point': '/run/fsys2'})
        self.addCleanupMount("/run/fsys2")
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 1), "fsys2")
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 3), "/run/fsys2")
        b.assert_pixels(self.card("Stratis filesystems"), "fsys-rows")
        self.assertEqual(self.inode(m.execute("findmnt -n -o SOURCE /run/fsys2").strip()),
                         self.inode("/dev/stratis/pool0/fsys2"))
        m.write("/run/fsys2/FILE", "Hello Stratis!")

        # Check that they have entries in fstab
        self.assertNotEqual(m.execute("grep /run/fsys1 /etc/fstab"), "")
        self.assertNotEqual(m.execute("grep /run/fsys2 /etc/fstab"), "")

        # Rename one filesystem
        self.click_card_row("Stratis filesystems", 1)
        b.click(self.card_desc_action("Stratis filesystem", "Name"))
        self.dialog({'name': "fsys1-renamed"})
        b.wait_text(self.card_desc("Stratis filesystem", "Name"), "fsys1-renamed")
        b.click(self.card_parent_link())
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1-renamed")

        # Destroy one filesystem
        self.click_card_row("Stratis filesystems", 1)
        self.click_card_dropdown("Stratis filesystem", "Delete")
        self.dialog_wait_open()
        b.assert_pixels("#dialog", "delete-fsys")
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_visible(self.card("Stratis filesystems"))
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys1-renamed"))

        # Unmount and remount the other filesystem
        self.click_dropdown(self.card_row("Stratis filesystems", 1), "Unmount")
        self.confirm()
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys2 (not mounted)")
        self.click_dropdown(self.card_row("Stratis filesystems", 1), "Mount")
        self.dialog({})
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys2")

        # Make a copy of the filesystem
        self.click_dropdown(self.card_row("Stratis filesystems", 1), "Snapshot")
        self.dialog_wait_open()
        self.dialog_set_val('name', 'fsys2-copy')
        self.dialog_set_val('mount_point', '/run/fsys2-copy')
        self.dialog_set_val('at_boot', 'never')
        b.assert_pixels("#dialog", "copy-fsys")
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 1), "fsys2-copy")
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 3), "/run/fsys2-copy")

        self.assertEqual("Hello Stratis!", m.execute("cat /run/fsys2-copy/FILE"))

        # Delete the copy
        self.click_card_row("Stratis filesystems", 2)
        self.click_card_dropdown("Stratis filesystem", "Delete")
        self.confirm()
        b.wait_visible(self.card("Stratis filesystems"))
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys2-copy"))

        # Make an unmounted copy of the filesystem
        self.click_dropdown(self.card_row("Stratis filesystems", 1), "Snapshot")
        self.dialog_wait_open()
        self.dialog_set_val('name', 'fsys2-copy')
        self.dialog_set_val('at_boot', 'never')
        self.dialog_apply_secondary()
        self.dialog_wait_close()
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 1), "fsys2-copy")
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 3), "(not mounted)")

        # Delete the copy
        self.click_card_row("Stratis filesystems", 2)
        self.click_card_dropdown("Stratis filesystem", "Delete")
        self.confirm()
        b.wait_visible(self.card("Stratis filesystems"))
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys2-copy"))

        # Create an unmounted filesystem
        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_set_val('name', 'fsys-unmounted')
        self.dialog_apply_secondary()
        self.dialog_wait_close()

        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys-unmounted")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "(not mounted)")

        # Delete the unmounted filesystem
        self.click_card_row("Stratis filesystems", 1)
        self.click_card_dropdown("Stratis filesystem", "Delete")
        self.confirm()
        b.wait_visible(self.card("Stratis filesystems"))
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys-unmounted"))

        # Add a data blockdev
        b.click(self.card_button("Stratis pool", "Add block device"))
        self.dialog_wait_open()
        self.dialog_apply()
        self.dialog_wait_error("disks", "At least one")
        self.dialog_set_val('disks', {dev_3: True})
        b.assert_pixels("#dialog", "add-disk")
        self.dialog_apply()
        self.dialog_wait_close()

        b.wait_visible(self.card_row("Stratis pool", name=dev_3))
        b.wait_in_text(self.card_desc("Stratis pool", "Usage"), "12 GB")

        # Add a cache blockdev
        b.click(self.card_button("Stratis pool", "Add block device"))
        self.dialog({'tier': "cache",
                     'disks': {dev_4: True}})
        b.wait_in_text(self.card_row("Stratis pool", name=dev_4), "cache")

        # Add a second cache blockdev, this uses a different code path
        b.click(self.card_button("Stratis pool", "Add block device"))
        self.dialog({'tier': "cache",
                     'disks': {dev_5: True}})
        b.wait_in_text(self.card_row("Stratis pool", name=dev_5), "cache")

        # Rename the pool
        b.click(self.card_desc_action("Stratis pool", "Name"))
        self.dialog({'name': "pool0-renamed"})
        b.wait_text(self.card_desc("Stratis pool", "Name"), "pool0-renamed")

        # Create another filesystem
        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog({'name': 'fsys3',
                     'mount_point': '/run/fsys3'})
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 1), "fsys3")
        b.wait_text(self.card_row_col("Stratis filesystems", 2, 3), "/run/fsys3")
        self.assertEqual(self.inode(m.execute("findmnt -n -o SOURCE /run/fsys3").strip()),
                         self.inode("/dev/stratis/pool0-renamed/fsys3"))

        # Destroy the pool
        self.click_card_dropdown("Stratis pool", "Delete")
        self.dialog_wait_open()
        b.assert_pixels('#dialog', "delete-pool")
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_visible(self.card("Storage"))
        b.wait_not_present(self.card_row("Storage", name="pool0-renamed"))

        # Check that the entries have disappeared from fstab
        self.assertEqual(m.execute("grep /run/fsys1 /etc/fstab || true"), "")
        self.assertEqual(m.execute("grep /run/fsys2 /etc/fstab || true"), "")
        self.assertEqual(m.execute("grep /run/fsys3 /etc/fstab || true"), "")

        m.execute("! findmnt /run/fsys1")
        m.execute("! findmnt /run/fsys2")
        m.execute("! findmnt /run/fsys2-copy")
        m.execute("! findmnt /run/fsys3")

        # https://bugzilla.redhat.com/show_bug.cgi?id=2183084
        # Do this assersion in the end so that the previous checks still run.
        # After the stratis pool is deleted we can't check this, so use the value from earlier.
        self.assertFalse(udisk_contains_stratis_private)

    @testlib.skipImage("Stratis too old", "rhel-8-*")
    def testAlerts(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = self.add_loopback_disk(PV_SIZE)
        dev_2 = self.add_loopback_disk(PV_SIZE)
        b.wait_visible(self.card_row("Storage", name=dev_1))
        b.wait_visible(self.card_row("Storage", name=dev_2))

        # Create an encrypted pool with two block devices
        self.click_devices_dropdown("Create Stratis pool")
        self.dialog_wait_open()
        self.dialog_set_val("encrypt_pass.on", val=True)
        self.dialog_set_val("passphrase", "foodeeboodeebar")
        self.dialog_set_val("passphrase2", "foodeeboodeebar")
        self.dialog_set_val("disks", {dev_1: True})
        self.dialog_set_val("disks", {dev_2: True})
        self.dialog_apply()
        self.dialog_wait_close()

        b.wait_visible(self.card_row("Storage", name="pool0"))
        b.wait_not_present(self.card_row("Storage", name="pool0") + " .ct-icon-exclamation-triangle")

        # Check that there is no alert on the details page
        self.click_card_row("Storage", name="pool0")
        b.wait_visible(self.card("Encrypted Stratis pool"))
        b.wait_not_present('.pf-v5-c-alert')

        m.execute(f"""
JSON=$(sudo cryptsetup token export --token-id=1 {dev_1} \
       | jq '.key_description = "stratis-1-key-no-other-is-the-same"')
sudo cryptsetup token remove --token-id=1 {dev_1}
echo $JSON | sudo cryptsetup token import --token-id=1 {dev_1}
systemctl restart stratisd
        """)

        b.go('#/')
        b.wait_visible(self.card_row("Storage", name="pool0") + " .ct-icon-exclamation-triangle")

        self.click_card_row("Storage", name="pool0")
        b.wait_visible('.pf-v5-c-alert:contains("This pool is in a degraded state")')

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_set_val("name", "fsys1")
        self.dialog_set_val("mount_point", "/run/fsys1")
        self.dialog_apply()
        self.dialog_wait_alert("Pool is in state NoRequests where this action cannot be performed until the issue is resolved manually")

    @testlib.nondestructive
    def testCli(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = self.add_loopback_disk(PV_SIZE)
        dev_2 = self.add_loopback_disk(PV_SIZE)
        b.wait_visible(self.card_row("Storage", name=dev_1))
        b.wait_visible(self.card_row("Storage", name=dev_2))

        # Create a pool outside of Cockpit
        m.execute(f"stratis pool create TEST1 {dev_1} {dev_2}")
        b.wait_visible(self.card_row("Storage", name="TEST1"))
        b.wait_in_text(self.card_row("Storage", name=dev_1), "Stratis block device")
        b.wait_in_text(self.card_row("Storage", name=dev_1), "TEST1")
        b.wait_in_text(self.card_row("Storage", name=dev_2), "Stratis block device")
        b.wait_in_text(self.card_row("Storage", name=dev_1), "TEST1")

        # Create two filesystems outside of Cockpit
        m.execute("stratis filesystem create TEST1 fsys1")
        b.wait_visible(self.card_row("Storage", name="fsys1"))
        m.execute("stratis filesystem create TEST1 fsys2")
        b.wait_visible(self.card_row("Storage", name="fsys2"))

        mount = f"{self.vm_tmpdir}/fsys1"

        # Mount externally, adjust fstab with Cockpit
        self.click_card_row("Storage", name="fsys1")
        m.execute(f"mkdir {mount}; mount /dev/stratis/TEST1/fsys1 {mount}")
        b.click(self.card_button("Stratis filesystem", f"Mount automatically on {mount} on boot"))
        b.wait_not_present(self.card_button("Stratis filesystem", f"Mount automatically on {mount} on boot"))
        self.assertIn("stratis-fstab-setup", m.execute(f"grep {mount} /etc/fstab"))

        # Unmount externally, adjust fstab with Cockpit
        m.execute(f"umount {mount}")
        b.click(self.card_button("Stratis filesystem", "Do not mount automatically on boot"))
        b.wait_not_present(self.card_button("Stratis filesystem", "Do not mount automatically on boot"))
        self.assertIn("noauto", m.execute(f"grep {mount} /etc/fstab"))

        # Destroy them outside of Cockpit
        b.click(self.card_parent_link())
        b.wait_visible(self.card("Stratis filesystems"))
        m.execute("stratis filesystem destroy TEST1 fsys1")
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys1"))
        m.execute("stratis filesystem destroy TEST1 fsys2")
        b.wait_not_present(self.card_row("Stratis filesystems", name="fsys2"))

        # Destroy the pool outside of Cockpit
        m.execute("stratis pool destroy TEST1")
        b.wait_in_text("main", "Not found")

        b.go("#/")
        b.wait_visible(self.card("Storage"))
        b.wait_not_present(self.card_row("Storage", name="TEST1"))

    def testOverprovisioning(self):
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = self.add_loopback_disk(PV_SIZE)
        b.wait_visible(self.card_row("Storage", name=dev_1))

        # Create a pool with overprov off
        self.dialog_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                               expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                               self.dialog_check({"name": "pool0"})),
                               values={
                                   "disks": {dev_1: True},
                                   "overprov.on": False
                               })
        self.click_card_row("Storage", name="pool0")

        overprov_toggle = "input[aria-label='Allow overprovisioning']"
        b.wait_visible(f"{overprov_toggle}:not(:checked)")

        # Create filesystem with defaults.  Initial virtual size
        # should be appropriate.

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_wait_val('set_custom_size.enabled', val=True)
        self.dialog_set_val('name', 'fsys1')
        self.dialog_apply_secondary()
        self.dialog_wait_close()

        # Snapshot should be disabled now

        sel = self.card_row("Stratis filesystems", name="fsys1")
        b.click(self.dropdown_toggle(sel))
        b.wait_visible(self.dropdown_action(sel, "Snapshot") + "[disabled]")
        b.wait_text(self.dropdown_description(sel, "Snapshot"),
                    "Not enough free space")
        b.click(self.dropdown_toggle(sel))

        # Switch overprov on

        b.click(overprov_toggle)
        b.wait_visible(f"{overprov_toggle}:checked")

        # Snapshot succeeds

        self.click_dropdown(self.card_row("Stratis filesystems", name="fsys1"), "Snapshot")
        self.dialog_wait_open()
        self.dialog_set_val('name', 'snap1')
        self.dialog_apply_secondary()
        self.dialog_wait_close()

        # Filesystem creation does not ask for file size by default

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_wait_val('set_custom_size.enabled', val=False)
        self.dialog_cancel()
        self.dialog_wait_close()

        # Turning overprov off should be disabled now

        b.wait_visible(overprov_toggle + "[disabled]")

        # Delete snapshot

        self.click_dropdown(self.card_row("Stratis filesystems", name="snap1"), "Delete")
        self.confirm()

        # Switch overprov off

        b.click(overprov_toggle)
        b.wait_visible(f"{overprov_toggle}:not(:checked)")

    def testFilesystemSizes(self):
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = self.add_loopback_disk(PV_SIZE)
        b.wait_visible(self.card_row("Storage", name=dev_1))

        # Create a pool
        self.dialog_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                               expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                               self.dialog_check({"name": "pool0"})),
                               values={"disks": {dev_1: True}})
        self.click_card_row("Storage", name="pool0")

        # Create filesystem with initial virtual size and a limit

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog_wait_open()
        self.dialog_set_val('name', 'fsys1')
        self.dialog_set_val('set_custom_size.enabled', val=True)
        self.dialog_set_val('size', 800)
        self.dialog_set_val('set_custom_limit.enabled', val=True)
        self.dialog_set_val('limit', 1600)
        self.dialog_apply_secondary()
        self.dialog_wait_close()

        self.click_card_row("Stratis filesystems", name="fsys1")

        b.wait_text(self.card_desc("Stratis filesystem", "Virtual size"), "800 MB")
        b.wait_text(self.card_desc("Stratis filesystem", "Virtual size limit"), "1.60 GB")

        # Raise the limit

        b.click(self.card_desc_action("Stratis filesystem", "Virtual size limit"))
        self.dialog_wait_open()
        self.dialog_wait_val('size_options.custom_limit', val=True)
        self.dialog_wait_val('limit', 1600)
        self.dialog_set_val('limit', 3200)
        self.dialog_apply()
        self.dialog_wait_close()

        b.wait_text(self.card_desc("Stratis filesystem", "Virtual size limit"), "3.20 GB")

        # Remove limit

        b.click(self.card_desc_action("Stratis filesystem", "Virtual size limit"))
        self.dialog_wait_open()
        self.dialog_set_val('size_options.custom_limit', val=False)
        self.dialog_apply()
        self.dialog_wait_close()

        b.wait_text(self.card_desc("Stratis filesystem", "Virtual size limit"), "none")


@testlib.skipImage("No Stratis", "debian-*", "ubuntu-*")
@testlib.skipImage("commit 817c957899a4 removed Statis 2 support", "rhel-8-*")
class TestStorageStratisReboot(storagelib.StorageCase):
    # LUKS uses memory hard PBKDF, 1 GiB is not enough; see https://bugzilla.redhat.com/show_bug.cgi?id=1881829
    provision = {
        "0": {"memory_mb": 1536}
    }

    def setUp(self):
        super().setUp()
        exe = self.machine.execute

        if self.image == "arch":
            # Arch Linux does not enable systemd units by default
            exe("systemctl enable --now stratisd")
            self.addCleanup(self.machine.execute, "systemctl disable --now stratisd")

    def testEncrypted(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        dev_2 = "/dev/sdb"
        m.add_disk("4G", serial="DISK2")
        b.wait_visible(self.card_row("Storage", name=dev_2))

        dev_3 = "/dev/sdc"
        m.add_disk("4G", serial="DISK3")
        b.wait_visible(self.card_row("Storage", name=dev_3))

        passphrase = "foodeeboodeebar"

        # Create an encrypted pool with a filesystem, but don't mount
        # it.  Cockpit will chose a key description for the pool and
        # we occupy its first choice in order to force Cockpit to use
        # something else.
        create_pool_key(m, "pool0", "not-the-passphrase")
        self.dialog_open_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                    expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                                    self.dialog_check({"name": "pool0"})))
        self.dialog_set_val("encrypt_pass.on", val=True)
        self.dialog_set_val("passphrase", passphrase)
        self.dialog_set_val("passphrase2", passphrase)
        self.dialog_set_val("disks", {dev_1: True})
        b.assert_pixels("#dialog", "create-encrypted-pool",
                        # The small checkbox ticks render inconsistently
                        ignore=["input[type=checkbox]"])
        self.dialog_apply()
        self.dialog_wait_close()
        m.execute("stratis key unset pool0")

        self.click_card_row("Storage", name="pool0")
        b.wait_visible(self.card("Encrypted Stratis pool"))

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog({'name': 'fsys1',
                     'mount_point': '/run/fsys1',
                     'at_boot': 'local'},
                    secondary=True)
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1 (not mounted)")

        # Check that it has an entry in fstab and that it is "noauto"
        self.assertIn("noauto", m.execute("grep /run/fsys1 /etc/fstab"))

        # Add a data blockdev
        b.click(self.card_button("Encrypted Stratis pool", "Add block device"))
        self.dialog_wait_open()
        self.dialog_set_val('disks', {dev_2: True})
        self.dialog_apply()
        self.dialog_wait_error("passphrase", "Passphrase cannot be empty")
        self.dialog_set_val('passphrase', passphrase)
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_in_text(self.card_row("Encrypted Stratis pool", name=dev_2), "data")

        # Change the passphrase
        b.wait_visible(self.card_desc("Encrypted Stratis pool", "Passphrase") + " button:contains(Remove):disabled")
        b.click(self.card_desc("Encrypted Stratis pool", "Passphrase") + " button:contains(Change)")
        self.dialog({'old_passphrase': passphrase,
                     'new_passphrase': "boodeefoodeebar",
                     'new_passphrase2': "boodeefoodeebar"})
        # do it again, with the old passphrase in the keyring
        create_pool_key(m, "pool0", "boodeefoodeebar")
        b.click(self.card_desc("Encrypted Stratis pool", "Passphrase") + " button:contains(Change)")
        self.dialog({'new_passphrase': passphrase,
                     'new_passphrase2': passphrase})
        m.execute("stratis key unset pool0")

        # Add a cache blockdev
        b.click(self.card_button("Encrypted Stratis pool", "Add block device"))
        self.dialog_wait_open()
        self.dialog_set_val('tier', "cache")
        self.dialog_set_val('disks', {dev_3: True})
        self.dialog_set_val('passphrase', passphrase)
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_in_text(self.card_row("Encrypted Stratis pool", name=dev_3), "cache")

        self.reboot()
        m.start_cockpit()
        b.relogin()
        b.enter_page("/storage")

        b.wait_visible(self.card("Stratis pool"))
        b.wait_in_text(self.card("Stratis pool"), "DISK1")
        b.wait_in_text(self.card("Stratis pool"), "DISK2")

        # Unlock the pool
        b.click(self.card_button("Stratis pool", "Start"))
        self.dialog_wait_open()
        self.dialog_set_val('passphrase', "wrong-passphrase")
        self.dialog_apply()
        with b.wait_timeout(60):
            b.wait_visible("#dialog .pf-v5-c-alert.pf-m-danger")
        self.dialog_set_val('passphrase', passphrase)
        self.dialog_apply()
        self.dialog_wait_close()
        b.wait_text(self.card_desc("Encrypted Stratis pool", "Name"), "pool0")

        # Mount the filesystem
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1 (not mounted)")
        self.click_dropdown(self.card_row("Stratis filesystems", 1), "Mount")
        self.dialog({})
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")

        # Reboot (this requires the passphrase)
        self.setup_systemd_password_agent(passphrase)
        self.reboot()
        m.start_cockpit()
        b.relogin()
        b.enter_page("/storage")
        b.wait_text(self.card_desc("Encrypted Stratis pool", "Name"), "pool0")

        # Filesystem should be mounted now
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")

        # Destroy the pool
        self.click_card_dropdown("Encrypted Stratis pool", "Delete")
        self.confirm()
        b.wait_visible(self.card("Storage"))
        b.wait_not_present(self.card_row("Storage", name="pool0"))

        # Check that the entry has disappeared from fstab
        self.assertEqual(m.execute("grep /run/fsys1 /etc/fstab || true"), "")

    def testReboot(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        # Create a pool
        self.dialog_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                               expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                               self.dialog_check({"name": "pool0"})),
                               values={"disks": {dev_1: True}})
        self.click_card_row("Storage", name="pool0")

        # Create a filesystems
        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog({'name': 'fsys1',
                     'mount_point': '/run/fsys1'})
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")

        self.reboot()
        m.start_cockpit()
        b.relogin()
        b.enter_page("/storage")

        # Filesystem should be mounted now
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")

    def testAtBoot(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        # Create a pool
        self.dialog_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                               expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                               self.dialog_check({"name": "pool0"})),
                               values={"disks": {dev_1: True}})
        self.click_card_row("Storage", name="pool0")

        def create(at_boot):
            b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
            self.dialog({'name': 'fsys1',
                         'mount_point': '/foo',
                         'at_boot': at_boot})
            b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
            b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/foo")

        def destroy():
            self.click_card_row("Stratis filesystems", 1)
            self.click_card_dropdown("Stratis filesystem", "Delete")
            self.dialog_wait_open()
            self.dialog_apply_with_retry("Device or resource busy")
            b.wait_visible(self.card("Stratis filesystems"))
            b.wait_not_present(self.card_row("Stratis filesystems", name="fsys1"))

        create("local")
        self.assertNotIn("noauto", m.execute("findmnt --fstab -n -o OPTIONS /foo"))
        destroy()

        create("nofail")
        self.assertIn("nofail", m.execute("findmnt --fstab -n -o OPTIONS /foo"))
        destroy()

        create("netdev")
        self.assertIn("_netdev", m.execute("findmnt --fstab -n -o OPTIONS /foo"))
        destroy()

        create("never")
        self.assertIn("x-cockpit-never-auto", m.execute("findmnt --fstab -n -o OPTIONS /foo"))
        self.assertIn("noauto", m.execute("findmnt --fstab -n -o OPTIONS /foo"))
        destroy()

    @testlib.skipImage("Stratis too old", "rhel-8-*")
    def testPoolResize(self):
        m = self.machine
        b = self.browser

        self.login_and_go("/storage")

        dev = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev))

        # Create a logical volume that we will later grow
        m.execute(f"vgcreate vgroup0 {dev}; lvcreate vgroup0 -n lvol0 -L 1500000256b")
        b.wait_visible(self.card_row("Storage", name="lvol0"))

        # Create a pool
        self.dialog_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                               expect=lambda: self.dialog_is_present('disks', "lvol0"),
                               values={"disks": {"lvol0": True}})
        b.wait_in_text(self.card_row("Storage", name="pool0"), "1.5 GB")

        # Grow the logical volume in Cockpit, the pool should grow automatically
        self.click_card_row("Storage", name="lvol0")
        b.click(self.card_button("LVM2 logical volume", "Grow"))
        self.dialog({"size": 1600})
        b.go("#/")
        b.wait_in_text(self.card_row("Storage", name="pool0"), "1.6 GB")

        # Grow the logical volume from outside of Cockpit, the pool should complain
        m.execute("lvresize vgroup0/lvol0 -L +100000256b")
        b.wait_visible(self.card_row("Storage", name="pool0") + ' .ct-icon-exclamation-triangle')
        self.click_card_row("Storage", name="pool0")
        b.wait_visible('.pf-v5-c-alert:contains("This pool does not use all the space")')
        b.click('button:contains("Grow the pool")')
        b.wait_not_present('.pf-v5-c-alert')
        b.wait_in_text(self.card_desc("Stratis pool", "Usage"), "1.7 GB")

        b.go("#/")

        # Grow the logical volume from outside of Cockpit, the logical volume should also complain
        m.execute("lvresize vgroup0/lvol0 -L +100000256b")
        b.wait_visible(self.card_row("Storage", name="lvol0") + " .ct-icon-exclamation-triangle")
        self.click_card_row("Storage", name="lvol0")
        # First shrink the volume to test whether Cockpit can figure out the right size for that
        b.click(self.card_button("LVM2 logical volume", "Shrink volume"))
        b.wait_in_text(self.card_desc("LVM2 logical volume", "Size"), "1.70 GB")
        b.wait_not_present(self.card_button("LVM2 logical volume", "Shrink volume"))
        # Then enlarge the volume from the outside again and grow the blockdev
        m.execute("lvresize vgroup0/lvol0 -L +100000256b")
        b.click(self.card_button("LVM2 logical volume", "Grow content"))
        b.wait_not_present(self.card_button("LVM2 logical volume", "Grow content"))
        b.go("#/")
        b.wait_in_text(self.card_row("Storage", name="pool0"), "1.8 GB")


@testlib.skipImage("No Stratis", "debian-*", "ubuntu-*", "arch")
class TestStoragePackagesStratis(packagelib.PackageCase, storagelib.StorageCase):

    def testStratisOndemandInstallation(self):
        m = self.machine
        b = self.browser

        # RHEL 8 should not offer installation of Stratis from Cockpit
        # itself.
        #
        ondemand_stratis = "rhel-8" not in m.image

        m.execute("systemctl stop stratisd && dnf remove -y stratisd stratis")
        if ondemand_stratis:
            self.addPackageSet("stratis")
            self.enableRepo()

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        if ondemand_stratis:
            self.click_devices_dropdown("Create Stratis pool")
            self.dialog_wait_open()
            b.wait_in_text("#dialog", "The stratisd package must be installed")
            self.dialog_apply()
            with b.wait_timeout(60):
                self.dialog_wait_val("name", "pool0")
            self.dialog_set_val("disks", {dev_1: True})
            self.dialog_apply()
            self.dialog_wait_close()
            b.wait_visible(self.card_row("Storage", name="pool0"))
        else:
            dropdown_toggle = self.dropdown_toggle(self.card_header("Storage"))
            raid_action = self.dropdown_action(self.card_header("Storage"), "Create MDRAID device")
            stratis_action = self.dropdown_action(self.card_header("Storage"), "Create Stratis pool")
            b.click(dropdown_toggle)
            b.wait_visible(raid_action)
            b.wait_not_present(stratis_action)


@testlib.skipImage("No Stratis", "debian-*", "ubuntu-*")
@testlib.skipImage("Stratis too old", "rhel-8-*")
class TestStorageStratisNBDE(packagelib.PackageCase, storagelib.StorageCase):
    provision = {
        "0": {"address": "10.111.112.1/20", "memory_mb": 2048},
        "tang": {"address": "10.111.112.5/20"}
    }

    def setUp(self):
        super().setUp()

        if self.image == "arch":
            # Arch Linux does not enable systemd units by default
            self.machine.execute("systemctl enable --now stratisd")
            self.addCleanup(self.machine.execute, "systemctl disable --now stratisd")

        self.stop_type_opt = get_stratis_stop_type_opt(self.machine.execute)

    def testBasic(self):
        m = self.machine
        b = self.browser

        tang_m = self.machines["tang"]
        tang_m.execute("systemctl start tangd.socket")
        tang_m.execute("firewall-cmd --add-port 80/tcp")

        self.login_and_go("/storage")

        dev_1 = "/dev/sda"
        m.add_disk("4G", serial="DISK1")
        b.wait_visible(self.card_row("Storage", name=dev_1))

        dev_2 = "/dev/sdb"
        m.add_disk("5G", serial="DISK2")
        b.wait_visible(self.card_row("Storage", name=dev_2))

        # Create an encrypted pool with both a passphrase and a keyserver
        self.dialog_open_with_retry(trigger=lambda: self.click_devices_dropdown("Create Stratis pool"),
                                    expect=lambda: (self.dialog_is_present('disks', dev_1) and
                                                    self.dialog_check({"name": "pool0"})))
        self.dialog_set_val("encrypt_pass.on", val=True)
        self.dialog_set_val("passphrase", "foodeeboodeebar")
        self.dialog_set_val("passphrase2", "foodeeboodeebar")
        self.dialog_set_val("encrypt_tang.on", val=True)
        self.dialog_set_val("tang_url", "10.111.112.5")
        self.dialog_set_val("disks", {dev_1: True})
        self.dialog_apply()
        b.wait_in_text("#dialog", "Check the key hash")
        b.wait_in_text("#dialog", tang_m.execute("tang-show-keys").strip())
        self.dialog_apply()
        with b.wait_timeout(60):
            self.dialog_wait_close()

        self.click_card_row("Storage", name="pool0")
        b.wait_visible(self.card_desc("Encrypted Stratis pool", "Passphrase"))
        b.wait_in_text(self.card_desc("Encrypted Stratis pool", "Keyserver"), "10.111.112.5")

        b.assert_pixels(self.card("Encrypted Stratis pool"), "header",
                        ignore=['.pf-v5-c-description-list__group:contains(UUID)'])

        # Remove passphrase
        b.click(self.card_desc("Encrypted Stratis pool", "Passphrase") + " button:contains(Remove)")
        self.confirm()
        b.wait_visible(self.card_desc("Encrypted Stratis pool", "Passphrase") + " button:contains(Add passphrase)")
        b.wait_visible(self.card_desc("Encrypted Stratis pool", "Keyserver") + " button:contains(Remove):disabled")

        # Stop the pool and start it again.  This should not ask
        # for the passphrase (since there isn't any)
        m.execute(f"stratis pool stop {self.stop_type_opt} pool0")
        b.wait_visible(self.card("Stratis pool"))
        tang_m.execute("systemctl stop tangd.socket")
        b.click(self.card_button("Stratis pool", "Start"))
        self.dialog_wait_open()
        # stratis' error message for unreachable tang server is very poor:
        # https://bugzilla.redhat.com/show_bug.cgi?id=2246920 Version < 3.6.0 said
        # "Error communicating with server 10.111.112.5", check this again after fixing
        b.wait_in_text("#dialog", "Error")
        self.dialog_cancel()
        self.dialog_wait_close()

        tang_m.execute("systemctl start tangd.socket")
        b.click(self.card_button("Stratis pool", "Start"))
        b.wait_visible(self.card("Encrypted Stratis pool"))

        # Put passphrase back and do the stopping starting again, but
        # without tangd running.  This should try clevis but then fall
        # back to asking for a passphrase.

        b.click(self.card_desc("Encrypted Stratis pool", "Passphrase") + " button:contains(Add passphrase)")
        self.dialog({'passphrase': "foodeeboodeebar",
                     'passphrase2': "foodeeboodeebar"})
        b.wait_visible(self.card_desc("Encrypted Stratis pool", "Passphrase") + " button:contains(Remove):not(:disabled)")
        m.execute(f"stratis pool stop {self.stop_type_opt} pool0")
        tang_m.execute("systemctl stop tangd.socket")
        b.click(self.card_button("Stratis pool", "Start"))
        self.dialog_wait_open()
        self.dialog_set_val("passphrase", "foobar")
        self.dialog_cancel()
        self.dialog_wait_close()

        # Finally start tang
        tang_m.execute("systemctl start tangd.socket")
        b.click(self.card_button("Stratis pool", "Start"))
        b.wait_visible(self.card("Encrypted Stratis pool"))

        # Add a blockdevice.  This requires the passphrase.

        b.click(self.card_button("Encrypted Stratis pool", "Add block device"))
        self.dialog({'disks': {dev_2: True}, 'passphrase': "foodeeboodeebar"})

        # Remove the keyserver and add it back

        b.click(self.card_desc("Encrypted Stratis pool", "Keyserver") + " button:contains(Remove)")
        self.confirm()

        b.click(self.card_desc("Encrypted Stratis pool", "Keyserver") + " button:contains(Add keyserver)")
        self.dialog_wait_open()
        self.dialog_set_val("tang_url", "10.111.112.5")
        self.dialog_set_val("passphrase", "foodeeboodeebar")
        self.dialog_apply()
        b.wait_in_text("#dialog", "Check the key hash")
        b.wait_in_text("#dialog", tang_m.execute("tang-show-keys").strip())
        self.dialog_apply()
        with b.wait_timeout(60):
            self.dialog_wait_close()
            b.wait_in_text(self.card_desc("Encrypted Stratis pool", "Keyserver"), "10.111.112.5")

        # Remove the keyserver and add it back a second time, but try
        # first with the wrong passphrase already in the keyring

        b.click(self.card_desc("Encrypted Stratis pool", "Keyserver") + " button:contains(Remove)")
        self.confirm()

        create_pool_key(m, "pool0", "foobar")
        b.click(self.card_desc("Encrypted Stratis pool", "Keyserver") + " button:contains(Add keyserver)")
        self.dialog_wait_open()
        self.dialog_set_val("tang_url", "10.111.112.5")
        self.dialog_apply()
        b.wait_in_text("#dialog", "Check the key hash")
        b.wait_in_text("#dialog", tang_m.execute("tang-show-keys").strip())
        self.dialog_apply()
        with b.wait_timeout(60):
            b.wait_in_text('#dialog', "Command failed")
        m.execute("stratis key unset pool0")
        create_pool_key(m, "pool0", "foodeeboodeebar")
        self.dialog_apply()
        with b.wait_timeout(60):
            self.dialog_wait_close()
            b.wait_in_text(self.card_desc("Encrypted Stratis pool", "Keyserver"), "10.111.112.5")
        m.execute("stratis key unset pool0")

        # Create a mounted filesystem and reboot.

        b.click(self.card_button("Stratis filesystems", "Create new filesystem"))
        self.dialog({'name': 'fsys1',
                     'mount_point': '/run/fsys1'})
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")
        self.reboot()
        m.start_cockpit()
        b.relogin()
        b.enter_page("/storage")
        b.wait_visible(self.card("Stratis pool"))
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 1), "fsys1")  # should be started after boot
        b.wait_text(self.card_row_col("Stratis filesystems", 1, 3), "/run/fsys1")  # should be mounted after boot


if __name__ == '__main__':
    testlib.test_main()
